package com.example.administrator.myapplication;

import android.content.Intent;
import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.view.View;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;


public class MainActivity extends AppCompatActivity implements View.OnClickListener {

	@Override
	protected void onCreate(Bundle savedInstanceState) {
		super.onCreate(savedInstanceState);
		setContentView(R.layout.activity_main);

		findViewById(R.id.base).setOnClickListener(this);
		findViewById(R.id.create).setOnClickListener(this);
		findViewById(R.id.just).setOnClickListener(this);
		findViewById(R.id.just_mutil_data).setOnClickListener(this);
		findViewById(R.id.from_array).setOnClickListener(this);
		findViewById(R.id.from_integer).setOnClickListener(this);
		findViewById(R.id.from_future_callback).setOnClickListener(this);
		findViewById(R.id.from_callback_time).setOnClickListener(this);

	}

	@Override
	public void onClick(View view) {
		switch (view.getId()) {
			case R.id.base:
				Intent intent = new Intent(this, BaseUseActivity.class);
				startActivity(intent);
				break;
			case R.id.create:
				Observable.create(new ObservableOnSubscribe<Integer>() {
					@Override
					public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
						try {
							if (!emitter.isDisposed()) {
								for (int i = 0; i < 10; i++) {
									emitter.onNext(i);
								}
								emitter.onComplete();
							}

						} catch (Exception e) {
							emitter.onError(e);
						}

					}
				}).subscribe(new Consumer<Integer>() {
					@Override
					public void accept(Integer integer) throws Exception {
						Log.d("zwk", "accept" + integer);
					}
				}, new Consumer<Throwable>() {
					@Override
					public void accept(Throwable throwable) throws Exception {
						Log.d("zwk", "throwabel" + throwable);
					}
				}, new Action() {
					@Override
					public void run() throws Exception {
						Log.d("zwk", "action");
					}
				});
				break;
			case R.id.just:
				Observable
						.just("hell world")
						.subscribe(new Consumer<String>() {
							@Override
							public void accept(String s) throws Exception {
								Log.d("zwk", s);
							}
						});
				break;

			case R.id.just_mutil_data:
				Observable
						.just("内容1", "内容2", "内容3")
						.subscribe(new Consumer<String>() {
							@Override
							public void accept(String s) throws Exception {
								Log.d("zwk", "接收到" + s);
							}
						}, new Consumer<Throwable>() {
							@Override
							public void accept(Throwable throwable) throws Exception {
								Log.d("zwk", "错误" + throwable);
							}
						}, new Action() {
							@Override
							public void run() throws Exception {
								Log.d("zwk", "action 完成的");
							}
						});

			case R.id.from_array:
				Observable
						.fromArray("array1", "array2")
						.subscribe(new Consumer<String>() {
							@Override
							public void accept(String s) throws Exception {
								Log.d("zwk", s);
							}
						});
				break;
			case R.id.from_integer:
				List<Integer> list = new ArrayList<>();
				for (int i = 0; i < 10; i++) {
					list.add(i);
				}
				Observable
						.fromIterable(list)
						.subscribe(new Consumer<Integer>() {
							@Override
							public void accept(Integer integer) throws Exception {
								Log.d("zwk", "接受到的内容" + integer);
							}
						}, new Consumer<Throwable>() {
							@Override
							public void accept(Throwable throwable) throws Exception {
								Log.d("zwk", "接受内容出错" + throwable);
							}
						}, new Action() {
							@Override
							public void run() throws Exception {
								Log.d("zwk", "action complement");
							}
						});
				break;

			case R.id.from_future_callback:
				ExecutorService service = Executors.newCachedThreadPool();
				Future<String> future = service.submit(new Callable<String>() {
					@Override
					public String call() throws Exception {
						Log.d("zwk", "开始模拟一些耗时的任务");
						Thread.sleep(50000);
						return "ok";
					}
				});
				Observable.fromFuture(future).subscribe(new Consumer<String>() {
					@Override
					public void accept(String s) throws Exception {
						Log.d("zwk", s);
					}
				});

				break;
			case R.id.from_callback_time:
				ExecutorService service1 = Executors.newCachedThreadPool();
				Future<String> future1 = service1.submit(new Callable<String>() {
					@Override
					public String call() throws Exception {
						Log.d("zwk", "模拟一些耗时的操作");
						Thread.sleep(5000);
						return "ok";

					}
				});
				Observable
						.fromFuture(future1, 4, TimeUnit.SECONDS)
						.subscribe(new Consumer<String>() {
							@Override
							public void accept(String s) throws Exception {
								Log.d("zwk", s);
							}
						}, new Consumer<Throwable>() {
							@Override
							public void accept(Throwable throwable) throws Exception {
								// 程序本应该报错，不过我抛出了异常
								Log.d("zwk", "报错的内容" + throwable.getMessage());
							}
						});
				break;
			default:
				break;
		}
	}
}
